package com.zz.common

import akka.kafka.ConsumerMessage.CommittableOffsetBatch
import cats.Functor
import com.zz.common.ExternalOffsetStore.TopicPartitionGroupIdOffset
import org.apache.kafka.clients.consumer.ConsumerRecord

sealed trait PassThroughT[+T, +P] {
  def data: T
  def offsets: P
}

sealed trait BatchOffsets
final case class BatchOffsetsA(offsets: Set[TopicPartitionGroupIdOffset]) extends BatchOffsets
final case class BatchOffsetsB(offsets: CommittableOffsetBatch)      extends BatchOffsets

final case class PassThrough[T](data: T, offsets: BatchOffsets) extends BatchOffsets

final case class PassA[+T](data: T, offsets: TopicPartitionGroupIdOffset) extends PassThroughT[T, TopicPartitionGroupIdOffset]

final case class PassB[+T](data: T, offsets: CommittableOffsetBatch) extends PassThroughT[T, CommittableOffsetBatch]

object PassThroughFunctor {

  implicit val passThroughAFunctor: Functor[PassB] = new Functor[PassB] {
    override def map[A, B](fa: PassB[A])(
        f: A => B
    ): PassB[B] = PassB(f(fa.data), fa.offsets)
  }

  implicit val passThroughBFunctor: Functor[PassA] = new Functor[PassA] {
    override def map[A, B](fa: PassA[A])(
        f: A => B
    ): PassA[B] = PassA(f(fa.data), fa.offsets)
  }

  implicit val passThroughFunctor: Functor[PassThrough] = new Functor[PassThrough] {
    override def map[A, B](fa: PassThrough[A])(
        f: A => B
    ): PassThrough[B] = PassThrough(f(fa.data), fa.offsets)
  }

}

object PassThrough {
  def fromKafkaConsumerRecord[V](groupId: String)(xs: Seq[ConsumerRecord[_, V]]): PassThrough[Seq[V]] = {
    val r: Set[TopicPartitionGroupIdOffset] = xs
      .map { x =>
        TopicPartitionGroupIdOffset(topic = x.topic, partition = x.partition, groupId = groupId, offset = x.offset, updatedAt = 0L)
      }
      .groupBy(_.topicPartitionGroupId)
      .map {
        case (k, v) =>
          val o = v.map(_.offset).max + 1L
          val r = TopicPartitionGroupIdOffset.fromTopicPartitionGroupId(k, o, System.currentTimeMillis())
          r
      }
      .toSet
    PassThrough(xs.map(_.value()), BatchOffsetsA(r))
  }
}
